package org.example.flink.connect;

import java.util.Date;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.example.flink.source.S1;
import org.example.flink.source.S2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import scala.Tuple3;

/**
 * Connection不同类型数据流合并; Union 相同类型的数据流合并
 */
public class FConnect {

    private static Logger logger = LoggerFactory.getLogger(FConnect.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Tuple3<String, Integer, Integer>> s1 = env.addSource(new S1());
        DataStreamSource<Tuple3<String, Integer, Integer>> s2 = env.addSource(new S2());

        DataStream<String> apply = s1.connect(s2)
            .map(new CoMapFunction<Tuple3<String, Integer, Integer>, Tuple3<String, Integer, Integer>, String>() {
                @Override
                public String map1(Tuple3<String, Integer, Integer> v1) throws Exception {
                    String s = new Date() + " " + v1._1() + " (" + v1._2() + "," + v1._3() + ")";
                    logger.info(s);
                    return s;
                }

                @Override
                public String map2(Tuple3<String, Integer, Integer> v2) throws Exception {
                    String s = new Date() + " " + v2._1() + " (" + v2._2() + "," + v2._3() + ")";
                    logger.info(s);
                    return s;
                }
            });

        env.execute("FConnect");

    }

}
